package com.shujia.streaming

import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo3ReduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {

    /**
     * 旧版本创建sparkContext的方式
     */
    //    val conf = new SparkConf()
    //    conf.setMaster("local[2]")
    //    conf.setAppName("spark streaming 单词统计案例1")
    //    val sparkContext = new SparkContext(conf)
    //    // def this(sparkContext: SparkContext, batchDuration: Duration) //传入一个spark core 环境对象以及一个接收时间的范围
    //    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    /**
     * 新版本中推荐使用SparkSession对象来获取
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local[2]")
      .appName("窗口演示")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()
    val sparkContext: SparkContext = sparkSession.sparkContext

    //创建SparkStreaming对象
    val streamingContext = new StreamingContext(sparkContext, Durations.seconds(5))

    val linesDS: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 12345)

    //将数据封装成kv格式，将来可以调用kv类型的算子来进行操作
    val wordsKVDS: DStream[(String, Int)] = linesDS.flatMap(_.split(" ")).map((_, 1))

    //1、如果只需要计算当前批次的数据，直接reduceByKey
    //2、如果要从最开始的数据计算，使用有状态算子，updateStateByKey和checkpoint配合使用
    //3、如果要计算最近的时间段内的数据，使用窗口类算子，reduceByKeyAndWindow

    /**
     * 滑动窗口
     */
//    val res1DS: DStream[(String, Int)] = wordsKVDS.reduceByKeyAndWindow(
//      (x: Int, y: Int) => x + y,
//      Durations.seconds(15), // 设置窗口的大小
//      Durations.seconds(5)) // 设置滑动的大小

    /**
     *  滚动窗口
     */
    val res1DS: DStream[(String, Int)] = wordsKVDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y,
      Durations.seconds(10), // 设置窗口的大小
      Durations.seconds(10)) // 设置滑动的大小

    res1DS.print()

    streamingContext.start()
    streamingContext.awaitTermination()
    streamingContext.stop()


  }
}
